package com.atguigu.edu.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.app.func.DimAsyncFunction;
import com.atguigu.edu.realtime.bean.CourseOrderBean;
import com.atguigu.edu.realtime.util.DateFormatUtil;
import com.atguigu.edu.realtime.util.MyClickhouseUtil;
import com.atguigu.edu.realtime.util.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.streaming.api.datastream.*;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.async.AsyncFunction;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author zhangsan
 * @Date 2022/10/20 15:01
 * @Description //TODO dws层 课程域 各类别各学科各课程下单情况
 */
public class DwsCourseOrderWindow {
    public static void main(String[] args) throws Exception{
        //TODO 1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(4);
        //TODO 2.检查点设置
        //TODO 3.从kafka 中读取数据 dwd_trade_order_detail
        String topic = "dwd_trade_order_detail";
        String groupId = "dws_course_order_window";
        FlinkKafkaConsumer<String> kafkaConsumer = MyKafkaUtil.getKafkaConsumer(topic, groupId);
        DataStreamSource<String> kafkaStrDS = env.addSource(kafkaConsumer);
        //kafkaStrDS.print();
        //TODO 4.数据类型转换 jsonStr->jsonObj 再转为实体类
        SingleOutputStreamOperator<JSONObject> jsonObjDS = kafkaStrDS.map(JSON::parseObject);
        SingleOutputStreamOperator<CourseOrderBean> orderBeanDS = jsonObjDS.map(
                new MapFunction<JSONObject, CourseOrderBean>() {
                    @Override
                    public CourseOrderBean map(JSONObject jsonObj) throws Exception {
                        String courseId = jsonObj.getString("course_id");
                        String courseName = jsonObj.getString("course_name");
                        String orderId = jsonObj.getString("order_id");
                        String userId = jsonObj.getString("user_id");
                        String finalAmounts = jsonObj.getString("final_amount");
                        Long ts = Long.valueOf(jsonObj.getString("ts"));

                        CourseOrderBean orderBean = CourseOrderBean.builder()
                                .courseId(courseId)
                                .courseName(courseName)
                                .orderId(orderId)
                                .userId(userId)
                                .finalAmountCnt(Double.valueOf(finalAmounts))
                                .ts(ts * 1000)
                                .orderIdSet(
                                        new HashSet<String>(
                                                Collections.singleton(orderId)
                                        )
                                )
                                .userIdSet(
                                        new HashSet<String>(
                                                Collections.singleton(userId)
                                        )
                                ).build();


                        return orderBean;
                    }
                }
        );
        //orderBeanDS.print("aaa");
        //TODO 5.指定watermark以及提取事件时间字段
        SingleOutputStreamOperator<CourseOrderBean> withWatermarkDS = orderBeanDS.assignTimestampsAndWatermarks(
                WatermarkStrategy.<CourseOrderBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        .withTimestampAssigner(
                                new SerializableTimestampAssigner<CourseOrderBean>() {
                                    @Override
                                    public long extractTimestamp(CourseOrderBean courseOrderBean, long l) {
                                        return courseOrderBean.getTs();
                                    }
                                }
                        )
        );
        //TODO 6.按照课程进行分组
        KeyedStream<CourseOrderBean, String> courseKeyedDS = withWatermarkDS.keyBy(CourseOrderBean::getCourseId);
        //TODO 7.开窗
        WindowedStream<CourseOrderBean, String, TimeWindow> windowDS = courseKeyedDS.window(TumblingEventTimeWindows.of(Time.seconds(10)));
        //TODO 8.聚合
        SingleOutputStreamOperator<CourseOrderBean> reduceDS = windowDS.reduce(
                new ReduceFunction<CourseOrderBean>() {
                    @Override
                    public CourseOrderBean reduce(CourseOrderBean value1, CourseOrderBean value2) throws Exception {
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        value1.getUserIdSet().addAll(value2.getUserIdSet());
                        value1.setFinalAmountCnt(value1.getFinalAmountCnt() + value2.getFinalAmountCnt());
                        return value1;
                    }
                },
                new WindowFunction<CourseOrderBean, CourseOrderBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow window, Iterable<CourseOrderBean> input, Collector<CourseOrderBean> out) throws Exception {
                        String stt = DateFormatUtil.toYmdHms(window.getStart());
                        String edt = DateFormatUtil.toYmdHms(window.getEnd());
                        for (CourseOrderBean orderBean : input) {
                            orderBean.setStt(stt);
                            orderBean.setEdt(edt);
                            orderBean.setOrderCnt((long) orderBean.getOrderIdSet().size());
                            orderBean.setUserCnt((long) orderBean.getUserIdSet().size());
                            orderBean.setTs(System.currentTimeMillis());
                            out.collect(orderBean);
                        }
                    }
                }
        );
        //reduceDS.print("rrr");
        //TODO 9.关联维度表 dim_course_info 取出subject_id
        SingleOutputStreamOperator<CourseOrderBean> withSubjectIdDS = AsyncDataStream.unorderedWait(
                reduceDS,
                new DimAsyncFunction<CourseOrderBean>("dim_course_info") {
                    @Override
                    public void join(CourseOrderBean orderBean, JSONObject dimInfoJsonObj) {
                        orderBean.setSubjectId(dimInfoJsonObj.getString("subject_id".toUpperCase()));
                    }

                    @Override
                    public String getKey(CourseOrderBean obj) {
                        return obj.getCourseId();
                    }
                }, 10*60, TimeUnit.SECONDS
        );
        //withSubjectIdDS.print("aa");

        //TODO 10.关联维度表 dim_base_subject_info 取出 subject_name  和 category_id
        SingleOutputStreamOperator<CourseOrderBean> withSubjectNameAndCategoryIdDS = AsyncDataStream.unorderedWait(
                withSubjectIdDS,
                new DimAsyncFunction<CourseOrderBean>("dim_base_subject_info") {
                    @Override
                    public void join(CourseOrderBean orderBean, JSONObject dimInfoJsonObj) {
                        orderBean.setSubjectName(dimInfoJsonObj.getString("subject_name".toUpperCase()));
                        orderBean.setCategoryId(dimInfoJsonObj.getString("category_id".toUpperCase()));
                    }

                    @Override
                    public String getKey(CourseOrderBean obj) {
                        return obj.getSubjectId();
                    }
                }, 10*60, TimeUnit.SECONDS
        );
        //withSubjectNameAndCategoryIdDS.print("aaa");
        //TODO 11.关联维度表 dim_base_category_info 取出 category_name
        SingleOutputStreamOperator<CourseOrderBean> withCategoryNameDS = AsyncDataStream.unorderedWait(
                withSubjectNameAndCategoryIdDS,
                new DimAsyncFunction<CourseOrderBean>("dim_base_category_info") {
                    @Override
                    public void join(CourseOrderBean orderBean, JSONObject dimInfoJsonObj) {
                        orderBean.setCategoryName(dimInfoJsonObj.getString("category_name".toUpperCase()));
                    }

                    @Override
                    public String getKey(CourseOrderBean obj) {
                        return obj.getCategoryId();
                    }
                }, 10*60, TimeUnit.SECONDS
        );
        //withCategoryNameDS.print("aaaa");
        //TODO 12.将聚合结果写到ck中
        withCategoryNameDS.print(">>>");
        withCategoryNameDS.addSink(MyClickhouseUtil.getSinkFunction("insert into dws_course_order_window values(?,?,?,?,?,?,?,?,?,?,?,?)"));

        env.execute();
    }
}
